package com.hugedata.idc.utils;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;

public class HDFSUtils
{

    public static void main(String[] args) throws IOException
    {
        String deleteSource = args[2];
        String srcFile = args[0];
        String destFile = args[1];
        if (StringUtils.equals("true", deleteSource))
        {
            move(srcFile, destFile);
        }
        else
        {
            copy(srcFile, destFile);
        }

    }

    public static int move(String srcFile, String destFile) throws IOException
    {
        Configuration conf = getConf();
        FileSystem fs = FileSystem.get(conf);
        Path dst = new Path(destFile);
        Path src = new Path(srcFile);
        if (fs.exists(src))
        {
            fs.rename(src, dst);
            return 1;
        }
        return 0;
    }

    public static int copy(String srcFile, String destFile) throws IOException
    {
        Configuration conf = getConf();
        FileSystem fs = FileSystem.get(conf);
        Path dst = new Path(destFile);
        Path src = new Path(srcFile);
        if (fs.exists(src))
        {
            FileUtil.copy(fs, src, fs, dst, false, conf);
            return 1;
        }
        return 0;
    }

    public static Configuration getConf()
    {
        Configuration config;
        String oozieProperty = System.getProperty("oozie.action.conf.xml");
        if (StringUtils.isNotEmpty(oozieProperty))
        {
            config = new Configuration(false);
            Path ooziePath = new Path("file:///", oozieProperty);
            config.addResource(ooziePath);
        }
        else
        {
            config = new Configuration();
        }
        return config;
    }

    public static int readfileFromHdfs(String srcFile, String destFile) throws IOException
    {

        Configuration conf = getConf();
        FileSystem fs = FileSystem.get(conf);
        try
        {
            Path dst = new Path(destFile);
            Path src = new Path(srcFile);
            if (fs.exists(src))
            {
                fs.copyToLocalFile(src, dst);
                return 1;
            }
        }
        finally
        {
            org.apache.hadoop.io.IOUtils.closeStream(fs);
        }
        return 0;
    }

    public static int readfileFromHdfsDir(String srcDir, String destLocalFile) throws IOException
    {
        Configuration conf = getConf();
        FileSystem fs = FileSystem.get(conf);
        Path src = new Path(srcDir);
        if (fs.exists(src))
        {
            String tmpPath = getHadoopTmpPath();
            Path dstFile = new Path(tmpPath, String.valueOf(System.currentTimeMillis()));
            FileUtil.copyMerge(fs, src, fs, dstFile, false, conf, "\n");
            Path dst = new Path(destLocalFile);
            fs.copyToLocalFile(dstFile, dst);
            return 1;
        }
        return 0;
    }

    /**
     * @return 作业运行时的临时目录
     */
    public static String getHadoopTmpPath()
    {
        Configuration conf = getConf();
        String hadoopTmpDir = conf.get("hadoop.tmp.dir");
        return hadoopTmpDir;
    }

    public static void merge(String srcFile, String destFile) throws IOException
    {
        Configuration conf = getConf();
        FileSystem fs = FileSystem.get(conf);
        Path src = new Path(srcFile);
        Path destPath = new Path(destFile);
        InputStream is = null;
        OutputStream os = null;
        try
        {
            is = fs.open(src);
            if (fs.exists(destPath))
            {
                os = fs.append(destPath);
            }
            else
            {
                os = fs.create(destPath);
            }
            IOUtils.copy(is, os);
        }
        finally
        {
            IOUtils.closeQuietly(is);
            IOUtils.closeQuietly(os);
        }
    }

}
